package com.tom.zojbackendjudgeservice.message;

import com.rabbitmq.client.Channel;
import com.tom.zojbackendcommon.common.ErrorCode;
import com.tom.zojbackendcommon.exception.BusinessException;
import com.tom.zojbackendjudgeservice.judge.JudgeService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;

import static com.tom.zojbackendcommon.constant.RabbitMqConstant.*;

/**
 * 消息消费者
 * 收到消息则将消息（题目提交id）送给判题服务
 */
@Component
@Slf4j
public class MessageConsumer {
    private int count = 0;

    @Resource
    private JudgeService judgeService;

        @RabbitListener(queues = QUEUE_NAME)
    public void receive(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long consumerTag)  {
        log.info("receive message: {}", message);
            if (message == null) {
                // 消息为空，则拒绝消息（不重试），进入死信队列
                try {
                    channel.basicNack(consumerTag, false, false);
                    throw new BusinessException(ErrorCode.PARAMS_ERROR, "消息为空");
                } catch (IOException e) {
                    throw new RuntimeException("拒绝消息失败！");
                }
            }

        // 发送消息给判题服务
        long questionSubmitId = Long.parseLong(message);
        judgeService.doJudge(questionSubmitId);
        try {
            //如果重试3次都失败，就不在把消息返回队列，直接将消息放入死信队列
            if(count >= 3) {
                channel.basicReject(consumerTag, false);
            }else{
                channel.basicAck(consumerTag, true);
                count++;
            }
        } catch (Exception e) {
            log.error("ack error", e);
            try {
                //没有接受到消息，就将消息重新放回队列中，在队列中重新消费
                channel.basicNack(consumerTag, false, false);
            } catch (IOException ex) {
                log.error("否定ack error", ex);
                throw new RuntimeException("否定ack失败！");
            }

        }
    }

    //死信队列
    @RabbitListener(queues = {QUEUE_NAME_DEAD})
    public void processMessageDead(String dataString, Message message, Channel channel) throws IOException {
        // 监听死信队列
        log.info("★[dead letter]dataString = " + dataString);
        log.info("★[dead letter]我是死信监听方法，我接收到了死信消息");
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

}